package com.shujia.flink.tf;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo05Reduce {
    public static void main(String[] args) throws Exception {

        // reduce：KeyBy之后的聚合
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);

        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);

        socketDS.flatMap((line, out) -> {
                    for (String word : line.split(",")) {
                        out.collect(word);
                    }
                }, Types.STRING)
                .map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0, Types.STRING)
                // 自定义聚合方式，也可以直接调用sum、max、min方法
                .reduce((v1, v2) -> {
                    // sum
                    return Tuple2.of(v1.f0, v1.f1 + v2.f1);
                }).print();

        env.execute();

    }
}
